package com.lmf.rabbitmq.four;

import com.lmf.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.util.Date;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 消息的发布确认：
 * 1.单个确认
 * 2.批量确认
 * 3.异步确认（推荐使用，但是比较复杂）
 */
public class Task03 {

    private final static int MaxCount = 1000;

    public static void main(String[] args) throws Exception {
        //1.单个确认
        publishMessageIndividually();   // 单个确认发布：确认1000条消息，需要749ms
        //2.批量确认
        publishMessageBatch();      //单个确认发布：确认1000条消息，需要170ms
        //3.异步确认
        publishMessageAsync();      //异步确认发布：确认1000条消息，需要59ms
                                    //异步确认发布：确认1000条消息，需要29ms
    }

    //单个确认发布
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();//生成队列名
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //获取当前系统时间戳
        long begin = System.currentTimeMillis();//开始时间

        //批量发送大量消息
        for (int i = 0; i < MaxCount; i++) {
            String message = i + "";
            //发送消息
            channel.basicPublish("", queueName, null, message.getBytes());
            //服务端返回 false 或超时时间内未返回，生产者可以消息重发
            boolean flag = channel.waitForConfirms();//单个确认
            if (flag) {
                System.out.println("消息发送成功");
            }
        }

        long end = System.currentTimeMillis();//结束时间戳
        System.out.println("单个确认发布：确认1000条消息，需要" + (end - begin) + "ms");
    }

    //批量确认发布
    public static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();//生成队列名
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();
        //获取当前系统时间戳
        long begin = System.currentTimeMillis();//开始时间

        //批量发布 批量确认
        int batchSize = 100;    //指定每隔多少次确认一次
        for (int i = 0; i < MaxCount; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //利用取模
            if (i % batchSize == 0) {
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println("确认发布成功！");
                }

            }
        }

        long end = System.currentTimeMillis();//结束时间戳
        System.out.println("批量确认发布：确认1000条消息，需要" + (end - begin) + "ms");
    }

    //异步确认发布
    public static void publishMessageAsync() throws Exception {
        //获取信道
        Channel channel = RabbitMqUtils.getChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();//生成队列名
        channel.queueDeclare(queueName, true, false, false, null);
        //开启发布确认
        channel.confirmSelect();


        /**
         * 线程安全有序的一个哈希表 适用于高并发的情况下
         * 1、轻松地将序号与消息进行关联
         * 2、轻松地批量删除，只要给到序号
         * 3、支持高并发
         */
        ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        /**
         * 消息发布成功的回调函数：
         *      1.deliveryTag：消息在队列中的标识
         *      2.multiple：是否为批量确认
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {  //监视发布成功的消息
            //2.删除确认发布成功的消息,剩下的就是未确认的消息
            if (multiple) {
                //如果是批量确认，则批量清除
                ConcurrentNavigableMap<Long, String> headMap = outstandingConfirms.headMap(deliveryTag);
                headMap.clear();    //清除
            }else {
                //非批量确认，就单个清除
                outstandingConfirms.remove(deliveryTag);
            }

            //打印发布成功的标识
            System.out.println("发布成功的标识：" + deliveryTag);
        };
        //消息发布失败的回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> { //监视发布失败的消息

            //3. 打印一下未确认的消息都有哪些
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布失败的标识：" + deliveryTag + "\n发布失败的消息内容：" + message);
        };

        /**
         *  配置一个异步监听器,需要手动实现两个接口：
         *      1.ackCallback：消息发布成功的回调函数
         *      2.nackCallback：消息发布失败的回调函数
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        //获取当前系统时间戳
        long begin = System.currentTimeMillis();//开始时间
        //批量发布1000条信息
        for (int i = 0; i < MaxCount; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());

            //1.利用线程安全有序的一个哈希表，记录发布的每一条消息：
            //getNextPublishSeqNo():获取信道里面下一条发布消息的顺序标识
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
        }

        long end = System.currentTimeMillis();//结束时间戳
        System.out.println("异步确认发布：确认1000条消息，需要" + (end - begin) + "ms");

        //关闭发送连接
//        channel.close();
    }
}
